6cd7cdf94c7dfddc3602c26bdec3fc401cd66cc4,core/client/src/main/java/alluxio/client/netty/NettyUnderFileSystemFileWriter.java,NettyUnderFileSystemFileWriter,write,#InetSocketAddress#number#number#number[]#number#number#,60
Before Change
int offset, int length) throws IOException {
SingleResponseListener listener = null;
try {
ChannelFuture f = mClientBootstrap.connect(address).sync();
LOG.debug("Connected to remote machine {}", address);
Channel channel = f.channel();
listener = new SingleResponseListener();
mHandler.addListener(listener);
channel.writeAndFlush(new RPCFileWriteRequest(ufsFileId, fileOffset, length,
new DataByteArrayChannel(source, offset, length)));
RPCResponse response = listener.get(NettyClient.TIMEOUT_MS, TimeUnit.MILLISECONDS);
channel.close().sync();
switch (response.getType()) {
case RPC_FILE_WRITE_RESPONSE:
After Change
}
@Override
public void write(InetSocketAddress address, long ufsFileId, long fileOffset, byte[] source,
int offset, int length) throws IOException {
SingleResponseListener listener = null;
Channel channel = null;
Metrics.NETTY_UFS_WRITE_OPS.inc();
try {
channel = BlockStoreContext.acquireNettyChannel(address, mClientBootstrap) ;
listener = new SingleResponseListener();
((ClientHandler) channel.pipeline().last()).addListener(listener);
ChannelFuture channelFuture = channel.writeAndFlush(
new RPCFileWriteRequest(ufsFileId, fileOffset, length,
new DataByteArrayChannel(source, offset, length))).sync();
if (channelFuture.isDone() && !channelFuture.isSuccess()) {
LOG.error("Failed to read ufs file from %s for ufsFilId %d with error %s.",
address.toString(), ufsFileId, channelFuture.cause());
throw new IOException(channelFuture.cause());
}
RPCResponse response = listener.get(NettyClient.TIMEOUT_MS, TimeUnit.MILLISECONDS);
switch (response.getType()) {
case RPC_FILE_WRITE_RESPONSE:
RPCFileWriteResponse resp = (RPCFileWriteResponse) response;
RPCResponse.Status status = resp.getStatus();
LOG.debug("status: {} from remote machine {} received", status, address);
if (status != RPCResponse.Status.SUCCESS) {
throw new IOException(ExceptionMessage.UNDER_FILE_WRITE_ERROR.getMessage(ufsFileId,
address, status.getMessage()));
}
break;
case RPC_ERROR_RESPONSE:
RPCErrorResponse error = (RPCErrorResponse) response;
throw new IOException(error.getStatus().getMessage());
default:
throw new IOException(ExceptionMessage.UNEXPECTED_RPC_RESPONSE
.getMessage(response.getType(), RPCMessage.Type.RPC_FILE_WRITE_RESPONSE));
}
} catch (Exception e) {
Metrics.NETTY_UFS_WRITE_FAILURES.inc();
throw new IOException(e);
} finally {
if (channel != null && listener != null) {